package com.xxx.ware.ware_manage_sys.bean.event;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xxx.ware.ware_manage_sys.bean.constant.MQTopicConst;
import com.xxx.ware.ware_manage_sys.bean.dto.OptRecordDto;
import com.xxx.ware.ware_manage_sys.bean.entity.records.WmsErrRecord;
import com.xxx.ware.ware_manage_sys.service.IErrRecordService;
import com.xxx.ware.ware_manage_sys.service.IOptRecordService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class EventConsumer implements MQTopicConst {

    @Autowired
    private IOptRecordService optRecordService;

    @Autowired
    private IErrRecordService errRecordService;

    @Value("${custom.retries}")
    private Integer retries;

    private static final String CONSUME_SUCCESS = "消息消费成功";

//    public Integer getRetries() {
//        return retries;
//    }

    @KafkaListener(topics = {OPT_RECORD, ERR_OPT_RECORD},
            groupId = "community-consumer-group")
    public void receiveOptRecordMsg(ConsumerRecord record, Acknowledgment ack){
        log.info("Kafka消费主题：{}，分区：{}，偏移量：{}，消息：{}",
                record.topic(),
                record.partition(),
                record.offset(),
                record.value());
        if (record == null || record.value() == null) {
            log.warn("消息内容为空，主题：{}，分区：{}，偏移量：{}", record.topic(), record.partition(), record.offset());
            ack.acknowledge();
            return;
        }
        try {
            //处理消息
            handleOptRecordMsg(record);
            log.info(CONSUME_SUCCESS + "，Kafka消费主题：{}，分区：{}，偏移量：{}，消息：{}",
                                        record.topic(),
                                        record.partition(),
                                        record.offset(),
                                        record.value());
        } catch (Exception e) {
            try {
                for (int i = 0; i < retries; i++) {
                    //每次处理消息都延时3s
                    try {
                        TimeUnit.SECONDS.sleep(3L);
                    } catch (InterruptedException interruptedException) {
                        log.warn("操作记录消费者消费异常，重试时阻塞出现异常，异常信息：{}", e.getMessage());
                    }
                    //处理消息
                    if (handleOptRecordMsg(record)){
                        log.info("重试消费，" + CONSUME_SUCCESS + "，Kafka消费主题：{}，分区：{}，偏移量：{}，消息：{}",
                                record.topic(),
                                record.partition(),
                                record.offset(),
                                record.value());
                        //消费者正常消费，终止循环
                        break;
                    }
                }
            } catch (Exception retryException) { //重试仍然异常，异常消息保存
                log.warn("操作记录消费者消费异常，重试处理消息异常，异常信息：{}", e.getMessage());
                //异常消息插入数据库
                OptRecordDto optRecordDto = parseOptRecordMsg(record);
                if (optRecordDto != null) {
                    WmsErrRecord wmsErrRecord = new WmsErrRecord();
                    wmsErrRecord.setRecordId(optRecordDto.getOptId());
                    wmsErrRecord.setOptModule(optRecordDto.getOptModule());
                    wmsErrRecord.setErrTopic(record.topic());
                    wmsErrRecord.setErrMsg(JSON.toJSONString(optRecordDto));
                    wmsErrRecord.setErrReason(retryException.getMessage());
                    errRecordService.saveErrRecord(wmsErrRecord);
                }
            }finally {
                //重试提交偏移量
                log.info("重试偏移量提交");
                ack.acknowledge();
            }
        } finally {
            //提交偏移量
            log.info("偏移量提交");
            ack.acknowledge();
        }
    }

    private Boolean handleOptRecordMsg(ConsumerRecord record) {
        try {
            OptRecordDto optRecordDto = parseOptRecordMsg(record);
            if (optRecordDto == null) {
                return false;
            }
            //正常消费成功，返回true
            return optRecordService.addOptRecord(optRecordDto);
        } catch (Exception e) {
            log.warn("处理消费消息异常，异常原因：" + e);
            throw e;
        }
    }

    private OptRecordDto parseOptRecordMsg(ConsumerRecord record){
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (StringUtils.isBlank(event.getMsg())) {
            log.warn("待处理消息为空，主题：{}，分区：{}，偏移量：{}",
                    record.topic(),
                    record.partition(),
                    record.offset());
            return null;
        }
        OptRecordDto optRecordDto = JSONObject.parseObject(event.getMsg(), OptRecordDto.class);
        log.info("解析后的日志记录：{}", optRecordDto);
        return optRecordDto;
    }
}
